package spring.cloud.tasks.client.item_task_processor.task_items_processor;

import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.client.item_task_processor.Task;
import spring.cloud.tasks.client.item_task_processor.TaskItemAndParameter;
import spring.cloud.tasks.client.item_task_processor.TaskProcessor;
import spring.cloud.tasks.client.item_task_processor.task_items_processor.task_data_processor.SingleTaskDataProcessor;
import spring.cloud.tasks.client.item_task_processor.task_items_processor.task_data_processor.TaskDataProcessor;

import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class SleepWhenNoDataModeTaskItemsProcessor<TaskDataId> implements TaskItemsProcessor {


    protected volatile TaskProcessor ruleTaskProcessor;
    protected volatile TaskDataProcessor<TaskDataId> taskDataProcessor;
    //
    protected volatile List<TaskDataId> taskDataIdList = Collections.synchronizedList(new ArrayList<TaskDataId>());
    protected volatile List<Object> runningTaskDataIdList = Collections.synchronizedList(new ArrayList<Object>());
    protected volatile List<TaskDataId> maybeRepeatTaskDataIdList = Collections.synchronizedList(new ArrayList<TaskDataId>());
    //
    volatile List<Thread> threadList = Collections.synchronizedList(new ArrayList<Thread>());
    volatile Task task;
    volatile Comparator<TaskDataId> taskDataIdComparator;
    volatile Statistics statistics;
    volatile Lock fetchSingleTaskDataLock = new ReentrantLock();
    volatile Lock fetchMultipleTaskDataLock = new ReentrantLock();
    volatile Lock loadTaskDataLock = new ReentrantLock();

    volatile boolean multipleTaskDataProcessor = false;
    volatile boolean isSleeping = false;
    volatile boolean destroy = false;

    public SleepWhenNoDataModeTaskItemsProcessor(//
                                                 TaskProcessor ruleTaskProcessor,//
                                                 TaskDataProcessor<TaskDataId> taskDataProcessor,//
                                                 Statistics statistics) throws Exception//
    {
        this.task = this.ruleTaskProcessor.getTask();
        this.ruleTaskProcessor = ruleTaskProcessor;
        this.taskDataProcessor = taskDataProcessor;
        this.taskDataIdComparator = new TaskDataIdComparator(this.taskDataProcessor.getComparator());
        if (this.taskDataProcessor instanceof SingleTaskDataProcessor<?>) {
            task.setMaxExecuteNum(1);
            multipleTaskDataProcessor = false;
        } else {
            multipleTaskDataProcessor = true;
        }
        this.statistics = statistics;
        //
        updateTaskItemCurrentProcessorProgressInfoAtTaskProgressThread();
        //
        for (int i = 0; i < ruleTaskProcessor.getThreadCountPerTaskProcessor(); i++) {
            this.newThread(i);
        }
    }

    public void updateTaskItemCurrentProcessorProgressInfoAtTaskProgressThread() {
        ruleTaskProcessor.updateTaskItemCurrentProcessorProgressInfoAtTaskProgressThread();
    }

    //
    private void newThread(int index) {
        String threadName = this.ruleTaskProcessor.getTaskProcessorInfo().getTaskId() + "-" + this.ruleTaskProcessor.getTaskProcessorInfo().getProcessorId() + "-" + index;
        SleepWhenNoDataModeTaskProcessingThread thread = new SleepWhenNoDataModeTaskProcessingThread(this);//共享信息
        thread.setName(threadName);
        thread.start();
        threadList.add(thread);
    }

    /**
     * 是否正在休眠
     */
    public boolean isSleeping() {
        return this.isSleeping;
    }
    //

    public boolean noTaskDatas() {
        return this.taskDataIdList.size() == 0 && this.runningTaskDataIdList.size() == 0;
    }

    public void destroy() throws Exception {
        this.destroy = true;
        this.taskDataIdList.clear();
    }

    public void clearTaskDataIdList() {
        this.taskDataIdList.clear();
        //
        ruleTaskProcessor.updateTaskItemCurrentProcessorProgressInfoAtTaskProgressThread();
        //
    }

    //


    public TaskDataId getTaskDataId() {
        fetchSingleTaskDataLock.lock();
        try {
            TaskDataId taskDataId = null;
            while (true) {
                if (this.taskDataIdList.size() == 0) {
                    return null;//at last if has the same task
                } else {
                    taskDataId = this.taskDataIdList.remove(0);
                    if (this.isRepeatTaskDataId(taskDataId) == false) {
                        return taskDataId;
                    }
                }
            }
        } finally {
            fetchSingleTaskDataLock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    protected boolean isRepeatTaskDataId(TaskDataId taskDataId) {
        if (this.maybeRepeatTaskDataIdList.size() == 0) {
            return false;
        } else {
            TaskDataId[] taskDataIds = (TaskDataId[]) this.maybeRepeatTaskDataIdList.toArray();
            for (int i = 0; i < taskDataIds.length; i++) {
                if (this.taskDataIdComparator.compare(taskDataId, taskDataIds[i]) == 0) {
                    this.maybeRepeatTaskDataIdList.remove(taskDataIds[i]);
                    return true;
                }
            }
            return false;
        }

    }

    @SuppressWarnings("unchecked")
    public TaskDataId[] getTaskDataIds() {
        fetchMultipleTaskDataLock.lock();
        try {
            if (this.taskDataIdList.size() == 0) {
                return null;
            }
            int minSizeBetweenTaskListSizeAndMaxExecuteNumber = taskDataIdList.size() > task.getMaxExecuteNum() ? task.getMaxExecuteNum() : taskDataIdList.size();//min

            List<TaskDataId> taskDataIdList = new ArrayList<TaskDataId>();
            int count = 0;
            TaskDataId taskDataId = null;
            while (count < minSizeBetweenTaskListSizeAndMaxExecuteNumber && ((taskDataId = this.getTaskDataId()) != null)) {
                taskDataIdList.add(taskDataId);
                count = count + 1;
            }
            if (taskDataIdList.size() == 0) {
                return null;
            } else {
                return taskDataIdList.toArray((TaskDataId[]) Array.newInstance(taskDataIdList.get(0).getClass(), 0));
            }
        } finally {
            fetchMultipleTaskDataLock.unlock();
        }
    }

    protected int getTaskDataIdsFromDb() throws InterruptedException {
        //
        updateTaskItemCurrentProcessorProgressInfoAtTaskProgressThread();
        //
        loadTaskDataLock.lock();
        try {
            if (this.taskDataIdList.size() > 0 || this.destroy) {
                return this.taskDataIdList.size();
            }

            if (this.task.getSleepTimeWhenNoData() > 0) {
                this.isSleeping = true;
                TimeUnit.SECONDS.sleep(task.getSleepTimeWhenNoData());
                this.isSleeping = false;
            }

            putRunningTaskDataIdListToMaybeRepeatTaskDataIdList();

            try {
                List<TaskItemAndParameter> taskItemAndParameterList = this.ruleTaskProcessor.getTaskItemAndParameterListUntilTaskDataListIsAllFinshed();

                if (taskItemAndParameterList.size() > 0) {
                    List<TaskDataId> taskDataList = this.taskDataProcessor.getTaskDataList(//
                            task.getTaskParameter(),//
                            taskItemAndParameterList,//
                            this.ruleTaskProcessor.getTaskItemListSizeInZooKeeper(),//
                            task.getFetchNum()//
                            //
                    );
                    ruleTaskProcessor.getTaskProcessorInfo().setLastFetchDataTime(new Timestamp(ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis()));
                    if (taskDataList != null) {
                        this.taskDataIdList.addAll(taskDataList);
                    }
                } else {
                    //...
                }
                addFetchNum(taskDataIdList.size());
                if (taskDataIdList.size() <= 0) {
                    /**
                     * 没有数据进行休眠【核心要点】
                     */
                    if (this.ruleTaskProcessor.isContinueWhenNoTaskDataIdList()) {
                        if (task.getSleepTimeWhenNoData() > 0) {
                            this.isSleeping = true;
                            Thread.sleep(task.getSleepTimeWhenNoData());
                            this.isSleeping = false;
                        }
                    }
                }
                return this.taskDataIdList.size();
            } catch (Throwable ex) {
                log.error("", ex);
            }
            return 0;
        } finally {
            loadTaskDataLock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public void putRunningTaskDataIdListToMaybeRepeatTaskDataIdList() {
        fetchSingleTaskDataLock.lock();
        try {
            this.maybeRepeatTaskDataIdList.clear();
            if (this.runningTaskDataIdList.size() == 0) {
                return;
            }
            Object[] objects = this.runningTaskDataIdList.toArray();
            for (int i = 0; i < objects.length; i++) {
                if (!this.multipleTaskDataProcessor) {
                    this.maybeRepeatTaskDataIdList.add((TaskDataId) objects[i]);
                } else {
                    TaskDataId[] taskDataIds = (TaskDataId[]) objects[i];
                    for (int j = 0; j < taskDataIds.length; j++) {
                        this.maybeRepeatTaskDataIdList.add(taskDataIds[j]);
                    }
                }
            }
        } finally {
            fetchSingleTaskDataLock.unlock();
        }
    }

    //
    public void addFetchNum(long num) {
        this.statistics.addFetchDataCount(1);
        this.statistics.addFetchTaskFetchNum(num);
    }

    public void addSuccessNum(long num, long spendTime) {
        this.statistics.addTaskSuccessNum(num);
        this.statistics.addSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime) {
        this.statistics.addTaskFailNum(num);
        this.statistics.addSpendTime(spendTime);
    }

    class TaskDataIdComparator implements Comparator<TaskDataId> {
        Comparator<TaskDataId> comparator;

        public TaskDataIdComparator(Comparator<TaskDataId> comparator) {
            this.comparator = comparator;
        }

        public int compare(TaskDataId leftTask, TaskDataId rightTask) {
            statistics.addOtherCompareCount(1);
            return this.comparator.compare(leftTask, rightTask);
        }

        public boolean equals(Object object) {
            return this.comparator.equals(object);
        }
    }


}
